package com.atguigu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.beans.DimJoinFunction;
import com.atguigu.gmall.realtime.common.GmallConfig;
import com.atguigu.gmall.realtime.utils.HbaseUtil;
import com.atguigu.gmall.realtime.utils.RedisUtil;
import io.lettuce.core.api.StatefulRedisConnection;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.hadoop.hbase.client.AsyncConnection;

import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

/**
 * @author Felix
 * @date 2023/7/18
 * 发送异步请求进行维度关联
 */
public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T,T> implements DimJoinFunction<T> {

    private StatefulRedisConnection<String,String> redisConnection;
    private AsyncConnection hbaseConnection;
    private String tableName;

    public DimAsyncFunction(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        redisConnection = RedisUtil.getAsyncRedisConnection();
        hbaseConnection = HbaseUtil.getAsyncConnection();
    }

    @Override
    public void close() throws Exception {
        RedisUtil.closeAsyncRedisConn(redisConnection);
        HbaseUtil.closeAsyncConnection(hbaseConnection);
    }

    @Override
    public void asyncInvoke(T obj, ResultFuture<T> resultFuture) throws Exception {
        //创建一个异步编排对象  supplyAsync执行CompletableFuture任务，支持返回值
        CompletableFuture.supplyAsync(
            new Supplier<JSONObject>() {
                @Override
                public JSONObject get() {
                    JSONObject dimJsonObj = RedisUtil.asyncGetDimInfo(redisConnection, tableName + ":" + getKey(obj));
                    return dimJsonObj;
                }
            }
            // thenApply方法表示，第一个任务执行完成后，执行第二个回调方法任务，会将该任务的执行结果，
            // 作为入参，传递到回调方法中，并且回调方法是有返回值的
        ).thenApplyAsync(
            new Function<JSONObject, JSONObject>() {
                @Override
                public JSONObject apply(JSONObject jsonObj) {
                    //判断是否从Redis中查到数据
                    if(jsonObj == null){
                        //缓存没有命中，发送请求到habse中查询
                        System.out.println(tableName + "  " + getKey(obj) + "  从 hbase 读取");
                        jsonObj = HbaseUtil.getDimInfoFromHbaseByAsync(hbaseConnection, GmallConfig.HBASE_NAMESPACE,tableName,getKey(obj));
                        //将从hbase中查询到的数据放到缓存中
                        RedisUtil.asyncWriteDim(redisConnection,tableName +":" +getKey(obj),jsonObj);
                    }else{
                        System.out.println(tableName + "  " + getKey(obj) + "  从 redis 读取");
                    }
                    return jsonObj;
                }
            }
        ).thenAcceptAsync(
            new Consumer<JSONObject>() {
                @Override
                public void accept(JSONObject dimJsonObj) {
                    if(dimJsonObj != null){
                        join(obj,dimJsonObj);
                    }
                    // 获取数据库交互的结果并发送给ResultFuture的回调函数
                    resultFuture.complete(Collections.singleton(obj));
                }
            }
        );
    }

    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) throws Exception {
        // 根据自己情况进行处理
        throw new RuntimeException("异步超时:一般是其他原因导致的异步超时, 请检查: \n" +
            "1.检查集群是否都 ok: hdfs redis hbase kafka. \n" +
            "2.检查下 redis 的配置 bind 0.0.0.0 \n" +
            "3.检查用到的维度表是否都在,并且每张表都数据. 最好通过 maxwell-bootstrap 同步一下 \n" +
            "4. 找我" +
            "");
    }
}
